/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.flume.sink;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.apache.flume.serialization.EventSerializerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class BranchIOManager {

	private static final Logger logger = LoggerFactory.getLogger(BranchIOManager.class);

	private long idays;
	private long seriesTimestamp;
	private AtomicInteger fileIndex;
	private File baseDirectory;

	private File currentFile;

	private OutputStream outputStream;
	private EventSerializer serializer;

	private static final String DAY_8_FORMAT = "yyyyMMdd";

	private static final String TIME_6_FORMAT = "HHmmss";

	private static final String DEFAULT_PATH = "/var/log/flume";

	private static long dayMillis = 3600000 * 24;

	public BranchIOManager() {
		seriesTimestamp = System.currentTimeMillis();
		fileIndex = new AtomicInteger();
	}

	public BranchIOManager(String confDefaultDirectory, String key) {
		
		String logPath = BranchPathConfig.getDirectory(key);
		
		if (logPath == null || logPath.trim().length() == 0) {
			if (confDefaultDirectory == null || confDefaultDirectory.trim().length() == 0) {
				logPath = DEFAULT_PATH;
				logger.warn("No configured path with key {}. Using module default path {}", key, DEFAULT_PATH);
			} else {
				logPath = confDefaultDirectory;
				logger.warn("No configured path with key {}. Using configured default path {}", key,
						confDefaultDirectory);
			}
		}

		baseDirectory = new File(logPath);

		if (!baseDirectory.exists()) {
			baseDirectory.mkdirs();
			logger.info("mkdir directory {}", baseDirectory);
		}

		idays = System.currentTimeMillis() / dayMillis;
		seriesTimestamp = System.currentTimeMillis();
		fileIndex = new AtomicInteger();
	}

	public File nextFile() {

		String fileName = to8Day(new Date()) + "-" + seriesTimestamp + "-" + fileIndex.incrementAndGet();
		currentFile = new File(baseDirectory, fileName);
		
		logger.debug("New file {}", currentFile);

		return currentFile;

	}

	public File getCurrentFile() {
		if (currentFile == null) {
			return nextFile();
		}
		return currentFile;
	}

	public boolean dayChanged() {
		return idays != (System.currentTimeMillis() / dayMillis);
	}

	public void write(Event event) throws IOException {
		serializer.write(event);
	}

	public void rotate() {
		if (currentFile != null) {
			if (dayChanged()) {
				idays = System.currentTimeMillis() / dayMillis;
				seriesTimestamp = System.currentTimeMillis();
				fileIndex = new AtomicInteger();
			}
		}
		currentFile = null;
	}

	public void open(String serializerType, Context serializerContext) throws IOException {

		File currentFile = getCurrentFile();
		try {

			outputStream = new BufferedOutputStream(new FileOutputStream(currentFile));
			serializer = EventSerializerFactory.getInstance(serializerType, serializerContext, outputStream);
			serializer.afterCreate();

		} catch (IOException e) {
			throw new IOException("Failed to open file " + currentFile, e);
		}
	}

	public void flush() throws IOException {
		this.serializer.flush();
		this.outputStream.flush();
	}

	public void close() throws IOException {
		if (outputStream != null) {
			logger.debug("Closing file {}", getCurrentFile());
			try {
				serializer.flush();
				serializer.beforeClose();
				outputStream.close();
			} catch (IOException e) {
				throw new IOException("Unable to close file " + getCurrentFile(), e);
			} finally {
				serializer = null;
				outputStream = null;
			}
		}
	}

	public boolean haveClosed() {
		return outputStream == null || serializer == null;
	}

	public static String to8Day(Date date) {
		return DateFormatUtils.format(date, DAY_8_FORMAT);
	}

	public static String to6Time(Date date) {
		return DateFormatUtils.format(date, TIME_6_FORMAT);
	}

	public File getBaseDirectory() {
		return baseDirectory;
	}

	public void setBaseDirectory(File baseDirectory) {
		this.baseDirectory = baseDirectory;
	}

	public long getSeriesTimestamp() {
		return seriesTimestamp;
	}

	public AtomicInteger getFileIndex() {
		return fileIndex;
	}

	public EventSerializer getSerializer() {
		return this.serializer;
	}

	public void setSerializer(EventSerializer serializer) {
		this.serializer = serializer;
	}

	public OutputStream getOutputStream() {
		return this.outputStream;
	}
	
	public static class BranchFileItem{
		private boolean updated = false;
		private String directory;
		
		public BranchFileItem(){}
		public BranchFileItem(boolean Updated, String Directory){
			this.updated = Updated;
			this.directory = Directory;
		}
		
		public boolean fileHasChanged(String filePath){
			return !filePath.equals(this.directory);
		}
		
		public boolean isUpdated() {
			return updated;
		}
		public void setUpdated(boolean updated) {
			this.updated = updated;
		}
		public String getDirectory() {
			return directory;
		}
		public void setDirectory(String directory) {
			this.directory = directory;
		}
	}

}
